package org.monkey.socket.plugin.provider;

import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.Message;
import org.monkey.socket.plugin.config.SocketConstant;
import org.monkey.socket.plugin.config.SocketMqConfig;
import org.monkey.socket.plugin.service.IMsgConsumer;
import org.springframework.beans.factory.InitializingBean;

/**
 * 对接中间件 消息消费的接口 搭配 IMsgProducer使用，如果需要弃用默认rocketMq消息发布订阅实现,那么只需重新实现IMsgProducer和IMsgConsumer并注入
 */
@Slf4j
public class DefaultIMsgConsumer implements IMsgConsumer, InitializingBean {

  private DefaultMQPushConsumer consumer;


  private MsgAcceptProvider msgAcceptProvider;
  private SocketMqConfig socketMqConfig;

  public DefaultIMsgConsumer(MsgAcceptProvider msgAcceptProvider, SocketMqConfig socketMqConfig) {
    this.msgAcceptProvider = msgAcceptProvider;
    this.socketMqConfig = socketMqConfig;
  }

  @Override
  public void afterPropertiesSet() {
    try {
      consumer = new DefaultMQPushConsumer(
          SocketConstant.SOCKET_GROUP + "-" + socketMqConfig.getProfile());
      consumer.setNamesrvAddr(socketMqConfig.getNameServerAddr());
      consumer.subscribe(
          SocketConstant.SOCKET_TOPIC + "-" + socketMqConfig.getProfile(),
          "*");
      //consumeMessageBatchMaxSize 当消息大于多少时 会对消息进行拆分 提交多线程处理 默认为1
      consumer.setConsumeMessageBatchMaxSize(1);
      consumer.registerMessageListener(
          (MessageListenerConcurrently)
              (messages, context) -> {
                try {
                  for (Message message : messages) {
                    msgAcceptProvider.onMessage(message.getBody());
                    log.info("socket msg consumer success...");
                  }
                  return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                  log.error("socket msg consumer error:", e);
                  log.error("it will be reconsume later...");
                  return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
              });
      consumer.setInstanceName(StrUtil.uuid().replace("-", ""));
      consumer.start();
      log.info("socket msg consumer init success:group={},topic={},server={}",
          SocketConstant.SOCKET_GROUP,
          SocketConstant.SOCKET_TOPIC + "-" + socketMqConfig.getProfile(),
          socketMqConfig.getNameServerAddr());
    } catch (Exception e) {
      log.error("socket msg consumer init failed：", e);
      log.error("socket msg consumer init failed:group={},topic={},server={}",
          SocketConstant.SOCKET_GROUP,
          SocketConstant.SOCKET_TOPIC + "-" + socketMqConfig.getProfile(),
          socketMqConfig.getNameServerAddr());
    }
  }


}
